/*
 * Copyright 2022 The Open Islands Authors. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.openislands.oi.util;

import com.google.common.base.Strings;
import org.openislands.oi.constant.MessageCodeEnum;
import org.openislands.oi.error.ExitCode;
import org.openislands.oi.error.LogicException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.lang.reflect.Field;
import java.nio.file.Path;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.function.Supplier;

public final class CommandUtils {
    private static final Logger log = LoggerFactory.getLogger(CommandUtils.class);

    /***
     * Process implement class name under unix system
     */
    private static final String UNIX_CLASS = "java.lang.UNIXProcess";

    /***
     * Process clone class for post-processing operations
     */
    private static final Process clone;

    /***
     * Specific implementation of post-processing classes in different environment
     */
    private static final ProcessHandler processHandler;

    /***
     * Filter condition for search process
     */
    private static final ThreadLocal<String> filterThreadLocal = new ThreadLocal<>();

    private static final ExecutorService threadPool = new ThreadPoolExecutor(
            2, 100, 30, TimeUnit.SECONDS, new SynchronousQueue<>()
    );

    static {
        clone = initCloneProcess();
        processHandler = UNIX_CLASS.equals(clone.getClass().getName()) ? new UnixHandler() : new ImplHandler();
    }

    private static Process initCloneProcess() {
        try {
            return Runtime.getRuntime().exec(new String[]{"java", "-version"}, null, null);
        } catch (IOException e) {
            log.error(e.getMessage(), e);
            System.exit(ExitCode.COMMAND_UTILS_ERROR.ordinal());
            return null;
        }
    }

    public static String startAsyncSubprocess(String[] command, Map<String, String> env, File dir, OnCommandExecOutputCallback callback) {
        return startSubprocess(command, env, dir, callback, ExceptConsumer.nothing());
    }

    public static String startSyncSubprocess(String[] command, Map<String, String> env, File dir, OnCommandExecOutputCallback callback) {
        return startSubprocess(command, env, dir, callback, Process::waitFor);
    }

    public static String startSyncSubprocess(String[] command, Map<String, String> env, File dir, OnCommandExecOutputCallback callback,
                                             long timeout, TimeUnit timeUnit) {
        return startSubprocess(command, env, dir, callback, (proc) -> {
            if (!proc.waitFor(timeout, timeUnit) && proc.isAlive()) {
                proc.destroy();
            }
        });
    }

    public static String startSubprocess(String[] command, Map<String, String> env, File dir,
                                         OnCommandExecOutputCallback callback, ExceptConsumer<Process> processConsumer) {
        try {
            log.info("start subprocess by command: {}", String.join(" ", command));
            log.info("subprocess environment:");
            env.forEach((k, v) -> log.info("{}: {}", k, v));
            Process proc = Runtime.getRuntime().exec(command, generateEnvArgs(env), dir);
            threadPool.submit(new OutputHandler(proc.getInputStream(), OutputHandler.OUTPUT_STREAM, callback));
            threadPool.submit(new OutputHandler(proc.getErrorStream(), OutputHandler.ERROR_STREAM, callback));
            processConsumer.accept(proc);
            log.info("start subprocess done");
            return getProcessUri(proc);
        } catch (Exception e) {
            callback.onSubprocessError(e.getMessage());
        }
        return "";
    }

    public static void startSyncLineSubprocess(String[] command, Consumer<String> lineConsumer, long timeout, TimeUnit timeUnit) {
        startSyncSubprocess(command, new HashMap<>(), null,
                new OnCommandExecOutputCallback() {
                    @Override
                    public void onSubprocessSuccess(String line) {
                        lineConsumer.accept(line);
                    }

                    @Override
                    public void onSubprocessError(String line) {
                    }
                }, timeout, timeUnit
        );
    }

    public static boolean stop(String processUri) {
        if (Strings.isNullOrEmpty(processUri)) {
            throw new LogicException(MessageCodeEnum.PARAMETER_ERROR, String.format("process uri %s is not valid", processUri));
        }
        return stop(getProcessPid(processUri));
    }

    public static boolean stop(long pid) {
        try {
            processHandler.stop(pid);
            return true;
        } catch (Exception e) {
            return false;
        }
    }

    public static boolean isAlive(String processUri) {
        if (Strings.isNullOrEmpty(processUri)) {
            throw new LogicException(MessageCodeEnum.PARAMETER_ERROR, String.format("process uri %s is not valid", processUri));
        }
        return isAlive(getProcessPid(processUri));
    }

    public static boolean isAlive(long pid) {
        try {
            return processHandler.isAlive(pid);
        } catch (Exception e) {
            return false;
        }
    }

    private static void mkdirs(Path workDir) throws IOException {
        if (workDir == null) {
            return;
        }
        File path = workDir.toFile();
        if (path.exists() && path.isFile()) {
            throw new RuntimeException(String.format("%s is file not dir", workDir.toString()));
        } else if (!path.exists() && !path.mkdirs()) {
            throw new IOException("mkdirs path " + workDir + " failed");
        }
    }

    private static String[] generateEnvArgs(Map<String, String> extraEnv) {
        if (extraEnv != null && !extraEnv.isEmpty()) {
            return extraEnv.entrySet().stream().map(e -> String.format("%s=%s", e.getKey(), e.getValue())).toArray(String[]::new);
        } else {
            return null;
        }
    }

    public static synchronized long getPid(Process process) {
        long pid = -1;
        try {
            pid = processHandler.getPid(process);
        } catch (Exception ignored) {
        }
        return pid;
    }

    private static String getProcessUri(Process proc) {
        return String.format("pid://%d", getPid(proc));
    }

    private static long getProcessPid(String uri) {
        return Long.parseLong(uri.replace("pid://", ""));
    }

    public static boolean isUnix() {
        return processHandler instanceof UnixHandler;
    }

    public static <Res> Res fromFilterCondition(String filter, Supplier<Res> supplier) {
        filterThreadLocal.set(filter);
        try {
            return supplier.get();
        } finally {
            filterThreadLocal.remove();
        }
    }

    public interface ProcessHandler {
        void stop(long pid) throws Exception;

        boolean isAlive(long pid) throws Exception;

        long getPid(Process process) throws Exception;
    }

    // java.lang.UNIXProcess
    public static class UnixHandler implements ProcessHandler {
        @Override
        public void stop(long pid) throws ReflectiveOperationException {
            synchronized (CommandUtils.clone) {
                String filter = CommandUtils.filterThreadLocal.get();
                if (StringUtils.isBlank(filter)) {
                    Class<? extends Process> aClass = CommandUtils.clone.getClass();
                    // hasExited
                    Field field = aClass.getDeclaredField("hasExited");
                    field.setAccessible(true);
                    field.set(CommandUtils.clone, false);
                    // pid
                    field = aClass.getDeclaredField("pid");
                    field.setAccessible(true);
                    // destroy process
                    field.set(CommandUtils.clone, (int) pid);
                    CommandUtils.clone.destroy();
                } else {
                    this.stopByFilter(filter);
                }
            }
        }

        private void stopByFilter(String filter) {
            String cmd = String.format("ps -ax | grep %s | grep -v grep | awk '{print $1}' | xargs kill -9", filter);
            log.info("stop cmd:{}", cmd);
            startSyncLineSubprocess(new String[]{"/bin/sh", "-c", cmd}, line -> {
            }, 3, TimeUnit.SECONDS);
        }

        @Override
        public boolean isAlive(long pid) {
            final StringBuffer stringBuffer = new StringBuffer();
            String pidStr = pid + "";
            String cmd = "ps -a | grep " + pidStr;
            startSyncLineSubprocess(new String[]{"/bin/sh", "-c", cmd}, line -> stringBuffer.append(line).append("\n"),
                    3, TimeUnit.SECONDS);

            for (String str : stringBuffer.toString().split("\n")) {
                if (str.startsWith(pidStr)) {
                    return true;
                }
            }
            return false;
        }

        @Override
        public long getPid(Process process) throws Exception {
            Field field = process.getClass().getDeclaredField("pid");
            field.setAccessible(true);
            return field.getLong(process);
        }
    }

    // java.lang.ProcessImpl
    public static class ImplHandler implements ProcessHandler {
        private void setPid(long pid) throws ReflectiveOperationException {
            Field field = CommandUtils.clone.getClass().getDeclaredField("handle");
            field.setAccessible(true);
            field.set(CommandUtils.clone, pid);
        }

        @Override
        public void stop(long pid) throws ReflectiveOperationException {
            synchronized (CommandUtils.clone) {
                setPid(pid);
                CommandUtils.clone.destroy();
            }
        }

        @Override
        public boolean isAlive(long pid) throws ReflectiveOperationException {
            synchronized (CommandUtils.clone) {
                setPid(pid);
                return CommandUtils.clone.isAlive();
            }
        }

        @Override
        public long getPid(Process process) throws Exception {
            Field field = process.getClass().getDeclaredField("handle");
            field.setAccessible(true);
            return field.getLong(process);
        }
    }

    public interface ExceptConsumer<T> {
        void accept(T t) throws Exception;

        static <T> ExceptConsumer<T> nothing() {
            return t1 -> {
            };
        }
    }

    public interface OnCommandExecOutputCallback {
        void onSubprocessSuccess(String line);

        void onSubprocessError(String line);
    }

    private static class OutputHandler implements Runnable {
        private static final int OUTPUT_STREAM = 0;
        private static final int ERROR_STREAM = 1;
        private final InputStream in;
        private final OnCommandExecOutputCallback callback;
        private final int type;

        public OutputHandler(InputStream in, int type, OnCommandExecOutputCallback callback) {
            this.in = in;
            this.type = type;
            this.callback = callback;
        }

        public int getType() {
            return type;
        }

        @Override
        public void run() {
            try (BufferedReader buff = new BufferedReader(new InputStreamReader(this.in))) {
                String line;
                while ((line = buff.readLine()) != null) {
                    if (this.callback != null) {
                        this.callback.onSubprocessSuccess(line);
                        /*if (this.type == OutputHandler.ERROR_STREAM) {
                            this.callback.onSubprocessError(line);
                        } else {
                            this.callback.onSubprocessSuccess(line);
                        }*/
                    }
                }
            } catch (IOException e) {
                log.error(e.getMessage(), e);
            }
        }
    }
}
